package hbasecli.hbase;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import hbasecli.exception.ConvertException;
import lombok.Getter;
import lombok.Setter;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Getter
@Setter
public class HbaseRecordInput {
    private String rowKey;
    private String rowKeyType;
    private Map<String, String> col2val = new HashMap<>();

    public HbaseRecord toRecord() {
        HbaseRecord record = new HbaseRecord();
        record.setRowKey(convertValue(this.rowKeyType, this.rowKey));
        List<HbaseRecord.HbaseRecordColumn> columns = new ArrayList<>();
        for (String col : col2val.keySet()) {
            String[] strings = col.split(":");
            String family = strings[0];
            String qualifier = strings[1];
            String dataType = strings[2];
            String originalValue = col2val.get(col);
            byte[] value;
            value = convertValue(dataType, originalValue);
            HbaseRecord.HbaseRecordColumn column = new HbaseRecord.HbaseRecordColumn();
            column.setFamily(family);
            column.setQualifier(qualifier);
            column.setDataType(dataType);
            column.setValue(value);
            columns.add(column);
        }
        record.setColumns(columns);
        return record;
    }

    private byte[] convertValue(String dataType, String originalValue) {
        byte[] value;
        try {
            if ("string".equals(dataType)) {
                value = Bytes.toBytes(originalValue);
            } else if ("integer".equals(dataType)) {
                value = Bytes.toBytes(Integer.parseInt(originalValue));
            } else if ("long".equals(dataType)) {
                value = Bytes.toBytes(Long.parseLong(originalValue));
            } else if ("float".equals(dataType)) {
                value = Bytes.toBytes(Float.parseFloat(originalValue));
            } else if ("double".equals(dataType)) {
                value = Bytes.toBytes(Double.parseDouble(originalValue));
            } else if ("short".equals(dataType)) {
                value = Bytes.toBytes(Short.parseShort(originalValue));
            } else if ("boolean".equals(dataType)) {
                value = Bytes.toBytes(Boolean.parseBoolean(originalValue));
            } else if ("hex".equals(dataType)) {
                String string = StrUtil.cleanBlank(originalValue).toLowerCase();
                if (!string.matches("[0123456789abcdef]+")) {
                    throw new ConvertException("无法正确解析" + originalValue);
                }
                if (string.length() % 2 != 0) {
                    throw new ConvertException(originalValue + "的有效字符个数不是偶数");
                }
                value = hex2bytes(string);
            } else {
                throw new ConvertException("不支持的存储类型");
            }
            return value;
        } catch (Exception e) {
            e.printStackTrace();
            throw new ConvertException(originalValue + "不能以" + dataType + "类型存储");
        }
    }

    public static byte[] hex2bytes(String string) {
        byte[] value;
        int length = string.length() / 2;
        value = new byte[length];
        for (int i = 0; i < length; i++) {
            String substr = string.substring(i * 2, (i + 1) * 2);
            int number = 0;
            for (int j = 0; j < substr.length(); j++) {
                number = number * 16;
                number = number + hexChar2int(substr.charAt(j));
            }
            value[i] = (byte) number;
        }
        return value;
    }

    public static int hexChar2int(char c) {
        if (c >= '0' && c <= '9') {
            return ((int) c) - ((int) '0');
        } else if (c >= 'A' && c <= 'F') {
            return ((int) c - (int) 'A') + 10;
        } else if (c >= 'a' && c <= 'f') {
            return ((int) c - (int) 'a') + 10;
        } else {
            throw new RuntimeException("Invalid char " + c);
        }
    }

    @Override
    public String toString() {
        return "HbaseRecordInput:" + JSONUtil.toJsonStr(this);
    }
}